package com.wtwd.campus.rocketmq;

import com.wtwd.campus.attendancesend.AttendanceSender;
import com.wtwd.campus.attendancesend.AttendanceSenderFactory;
import com.wtwd.campus.common.Constants;
import com.wtwd.campus.config.ConsumerConfig;
import com.wtwd.campus.config.DefaultConsumerConfigure;
import com.wtwd.campus.entity.AttendanceCommand;
import com.wtwd.campus.entity.Command;
import com.wtwd.campus.entity.MqDeleteRedisCommand;
import com.wtwd.campus.entity.MqMessage;
import com.wtwd.campus.netty.ChannelMap;
import com.wtwd.campus.send.Sender;
import com.wtwd.campus.send.SenderFactory;
import com.wtwd.campus.service.IRedisService;
import com.wtwd.campus.utils.ByteObjectUtils;
import io.netty.channel.Channel;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;

import java.util.List;

@Configuration
public class ReceiveRocketMqMessage extends DefaultConsumerConfigure implements ApplicationListener<ApplicationReadyEvent> {

    Logger log = LoggerFactory.getLogger(ReceiveRocketMqMessage.class);

    @Autowired
    private ConsumerConfig consumerConfig;

    @Autowired
    private IRedisService redisService;


    @Override
    public void onApplicationEvent(ApplicationReadyEvent arg0) {
        try {
            super.listener(consumerConfig.getDeviceTopic(), consumerConfig.getDeviceTag());
        } catch (MQClientException e) {
            log.error("消费者监听器启动失败", e);
        }

    }

    @Override
    public ConsumeConcurrentlyStatus dealBody(List<MessageExt> msgs) {
        for (MessageExt msg : msgs) {
            try {
                MqMessage mqMessage = (MqMessage) ByteObjectUtils.toObject(msg.getBody());
                if ((Constants.CLOSE_CHANNEL).equals(mqMessage.getType())) {
                    //关闭通道
                    log.info("关闭通道：" + mqMessage.getType());
                    String imei = (String) mqMessage.getObject();
                    Channel channel = ChannelMap.get(imei);
                    if (channel != null) {
                        channel.close();
                        ChannelMap.removeByKey(imei);
                    }
                    continue;
                }
                log.info("接收到队列消息：" + mqMessage.toString());
                if (Constants.IDENTITY.equals(mqMessage.getType()) || Constants.HEART_BEAT.equals(mqMessage.getType()) || Constants.SEND_DATA.equals(mqMessage.getType()) || Constants.TIME_CHECK.equals(mqMessage.getType())) {
                    //考勤机指令下发
                    AttendanceSender sender = AttendanceSenderFactory.createSender(mqMessage);
                    log.info("选择考勤机Sender :" + sender);
                    AttendanceCommand command = sender.dealMessage(mqMessage);
                    boolean send = sender.send(command);
                    log.info("考勤机[{}]消息发送 {}", command.getContent(), send ? "成功" : "失败");
                } else if (Constants.DELETE_REDIS_COMMAND.equals(mqMessage.getType())) {
                    MqDeleteRedisCommand deleteRedisCommand = (MqDeleteRedisCommand) mqMessage.getObject();
                    redisService.deleteCommand(deleteRedisCommand.getImei(),deleteRedisCommand.getKey());
                } else {
                    Sender sender = SenderFactory.createSender(mqMessage);
                    log.info("选择Sender :" + sender);
                    Command command = sender.dealMessage(mqMessage);
                    boolean send = sender.send(command);
                    log.info("设备[{}]消息发送 {}", command.getContent(), send ? "成功" : "失败");
                }
            } catch (Exception e) {
                e.printStackTrace();
                log.error("body转字符串解析失败-----------", e);
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
